package cn.com.sparkle.paxos.net.netlayer.jvmpipe;

import java.util.concurrent.ArrayBlockingQueue;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.checksum.ChecksumUtil;
import cn.com.sparkle.paxos.net.frame.FrameBody;
import cn.com.sparkle.paxos.net.netlayer.NetCloseException;
import cn.com.sparkle.paxos.net.netlayer.NetHandler;
import cn.com.sparkle.paxos.net.netlayer.PaxosSession;

public class JvmPipePaxosSession extends PaxosSession {

	private final static Logger logger = Logger.getLogger(JvmPipePaxosSession.class);
	private volatile boolean isClose = false;

	private NetHandler handler;
	private JvmPipePaxosSession peer;
	private ArrayBlockingQueue<FrameBody> queue = new ArrayBlockingQueue<FrameBody>(1000);
	private Thread recieveThread;

	public JvmPipePaxosSession(NetHandler handler, String flag) {
		this.handler = handler;
		recieveThread = new Thread() {
			public void run() {
				while (true) {
					try {
						FrameBody body = queue.take();
						JvmPipePaxosSession.this.handler.onRecieve(JvmPipePaxosSession.this, body);
					} catch (InterruptedException e) {
						// close elegantly
					} catch (Throwable e) {
						logger.error("error", e);
						closeSession();
					}
				}
			}
		};
		recieveThread.setName("pipe-injvm" + flag);
		recieveThread.start();
	}

	@Override
	public void write(FrameBody message) throws NetCloseException {
		synchronized (this) {
			if(isClose){
				throw new NetCloseException(null);
			}
			peer.queue.add(message);
		}
	}

	public void setPeer(JvmPipePaxosSession peer) {
		this.peer = peer;
	}

	@Override
	public void closeSession() {
		synchronized (this) {
			if (!isClose) {
				recieveThread.interrupt();
				isClose = true;
				Thread t = new Thread() {
					public void run() {
						//async to trigger disconnect event
						handler.onDisconnect(JvmPipePaxosSession.this);
					}
				};
				t.setName("jvmpipe-close-thread");
				t.start();

			} else {
				return;
			}
		}
		peer.closeSession();
	}

	@Override
	public int getChecksumType() {
		return ChecksumUtil.NO_CHECKSUM;
	}

}
